package de.tuberlin.dima.minidb.impl.tablePage;

import de.tuberlin.dima.minidb.catalogue.ColumnSchema;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import de.tuberlin.dima.minidb.catalogue.TableSchema;
import de.tuberlin.dima.minidb.core.BigIntField;
import de.tuberlin.dima.minidb.core.CharField;
import de.tuberlin.dima.minidb.core.DataField;
import de.tuberlin.dima.minidb.core.DataTuple;
import de.tuberlin.dima.minidb.core.DataType;
import de.tuberlin.dima.minidb.core.DateField;
import de.tuberlin.dima.minidb.core.DoubleField;
import de.tuberlin.dima.minidb.core.FloatField;
import de.tuberlin.dima.minidb.core.IntField;
import de.tuberlin.dima.minidb.core.TimeField;
import de.tuberlin.dima.minidb.core.TimestampField;
import de.tuberlin.dima.minidb.io.PageFormatException;
import de.tuberlin.dima.minidb.io.tables.PageTupleAccessException;
import de.tuberlin.dima.minidb.io.tables.PageTupleIterator;
import de.tuberlin.dima.minidb.io.tables.TablePage;

public class TablePageImpl implements TablePage{
	private TableSchema schema;
	private byte[] binaryPage;

	
	public TablePageImpl(TableSchema schema, byte[] binaryPage) {
		this.schema = schema;
		this.binaryPage = binaryPage;
	}

	/*
	 *  init TablePage aka. Block 
	 */
	public TablePageImpl(TableSchema schema, byte[] binaryPage, int newPageNumber) {
		this.schema = schema;
		this.binaryPage = binaryPage;
		
		// set magic Number
		setMagicNumber();
		
		// set PageNumber
		setPageNumber(newPageNumber);
		
		// init number of tuples aka. record
		setNumberOfTuples(0);
		
		// set tuple width
		setTupleWidth();
		
		// set variable-length chunks offset
		setVariableOffset(schema.getPageSize());
	}
	/* Header Legends
	 * 0-3   : Magic Number
	 * 4-7   : Page Number
	 * 8-11  : Number Of Tuples
	 * 12-15 : Tuple-Width
	 * 16-19 : Variable-Length Chunk Offset
	 */
	
	/*
	 * Tuple = Record 
	 */
	
	/*
	 * Vicky
	 */
	private void setVariableOffset(int offset){
		int i = offset;
		
		for (int j = 0; j < 4; j++) {
			binaryPage[16+j] = (byte)( Integer.reverse( i>>> 8*j & 0xff  ) >>> 24);
		}
	}
	
	/*
	 * Vicky
	 */
	private int getVariableOffset(){
		int offset=0;
		for (int j = 0; j < 4; j++) {
			offset |= (Integer.reverse(binaryPage[j+16] &0xff ) >>> (3-j)*8);
			}
		return offset;
	}
	
	/*
	 * calculate new Variable Offset and write to header. aka. update variable offset info
	 * @author Vicky
	 */
	private void updateVariableOffset(int shift) {
		int newOffset = getVariableOffset() -  shift;
		setVariableOffset(newOffset);
	}
	
	/*
	 *  used once when constructor is called.
	 */
	private void setTupleWidth() {
		int recordWidth = 0;
		for (int i = 0; i < schema.getNumberOfColumns(); i++) {
			recordWidth += (schema.getColumn(i).getDataType().isFixLength()?
					schema.getColumn(i).getDataType().getNumberOfBytes():DataType.BIG_INT.getNumberOfBytes());
		}
		for (int j = 0; j < 4; j++) {
			binaryPage[12+j] = (byte)( Integer.reverse( recordWidth >>> 8*j & 0xff  ) >>> 24);
		}
	}
	/*
	 * Vicky
	 */
	private void setNumberOfTuples(int i) {
		int numberOfTuples = i;
		
		for (int j = 0; j < 4; j++) {
			binaryPage[8+j] = (byte)( Integer.reverse( numberOfTuples>>> 8*j & 0xff  ) >>> 24);
		}
		
	}

	private void setMagicNumber() {
		int i = 0xdeadbeef;
		
		for (int j = 0; j < 4; j++) {
			binaryPage[j] = (byte)( Integer.reverse( i>>> 8*j & 0xff  ) >>> 24);
		}
	}

	/*
	 * set page number in binaryPage
	 */
	private void setPageNumber(int newPageNumber) {
		for (int j = 4; j < 8; j++) {
			binaryPage[j] = (byte)( Integer.reverse( newPageNumber >>> 8*j & 0xff  ) >>> 24);
		}
	}

	/*
	 * delete position-th record, start with 0.
	 * @see de.tuberlin.dima.minidb.io.tables.TablePage#deleteDataTuple(int)
	 * @author thuy
	 */
	@Override
	public void deleteDataTuple(int position) throws PageTupleAccessException {
		int recordOffset = 5 + (position) * getRecordWidth();
		new IntField(1).encodeBinary(binaryPage, recordOffset);
	}

	/*
	 * Vicky
	 */
	/*
	 * if e.g. position = 5, we must just spring 5 :)) DataTuples
	 */
	@Override
	public DataTuple getDataTuple(int position) throws PageTupleAccessException {
		int offset = 32 + (position) * getRecordWidth();
		if (isTombstone(offset+3))
		{
			throw new PageTupleAccessException(1, "getDataTuple -> Tuple is already deleted");
		}
		else
		{
			DataField[] dataField = new DataField[schema.getNumberOfColumns()];
			offset += 4;
			for (int i = 0; i < schema.getNumberOfColumns(); i++)
			{
				ColumnSchema cs = schema.getColumn(i);
				DataType dt = cs.getDataType();
				if (dt == DataType.BIG_INT)
				{
					BigIntField bigintf = BigIntField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = bigintf;
					offset += DataType.BIG_INT.getNumberOfBytes();
				}
				else if (dt == DataType.CHAR)
				{
					// read 1 Byte
					CharField charf = CharField.getFieldFromBinary(this.binaryPage, offset, 1);
					dataField[i] = charf;
					offset += DataType.CHAR.getNumberOfBytes();
				}
				else if (dt == DataType.DATE)
				{
					DateField datef = DateField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = datef;
					offset += DataType.DATE.getNumberOfBytes();
				}
				else if (dt == DataType.DOUBLE)
				{
					DoubleField doublef = DoubleField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = doublef;
					offset += DataType.DOUBLE.getNumberOfBytes();
				}
				else if (dt == DataType.FLOAT)
				{
					FloatField floatf = FloatField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = floatf;
					offset += DataType.FLOAT.getNumberOfBytes();
				}
				else if (dt == DataType.INT)
				{
					IntField intf = IntField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = intf;
					offset += DataType.INT.getNumberOfBytes();
				}
				else if (dt == DataType.SMALL_INT)
				{
					FloatField floatf = FloatField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = floatf;
					offset += DataType.SMALL_INT.getNumberOfBytes();
				}
				else if (dt == DataType.TIME)
				{
					TimeField timef = TimeField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = timef;
					offset += DataType.TIME.getNumberOfBytes();
				}
				else if (dt == DataType.TIMESTAMP)
				{
					TimestampField timestampf = TimestampField.getFieldFromBinary(this.binaryPage, offset);
					dataField[i] = timestampf;
					offset += DataType.TIMESTAMP.getNumberOfBytes();
				}
				else if (dt == DataType.VAR_CHAR)
				{
					BigIntField bigintf2 = BigIntField.getFieldFromBinary(this.binaryPage, offset);
					int start  = (int) (bigintf2.getValue() & 0xffffffff);
					int length = (int) (bigintf2.getValue() >>> 32);
					CharField charf2 = CharField.getFieldFromBinary(this.binaryPage, start, length);
					dataField[i] = charf2;
					offset += length;
				}
			}
			return new DataTuple(dataField);
		}
	}
	
	/*
	 * Vicky
	 */
	private boolean isTombstone(int offset){
		if ( (binaryPage[offset] & 0x01) == 0x01)
			return true;
		else
			return false;
	}

	
	@Override
	public PageTupleIterator getIterator() throws PageTupleAccessException {
		// TODO Auto-generated method stub
		
		return new PageTupleIterator() {
			int visitedRecordNum = 0;
			@Override
			public boolean hasNext() throws PageTupleAccessException {
				// TODO Auto-generated method stub
				return (visitedRecordNum >= getNumRecordsOnPage())?false:true;
			}

			@Override
			public DataTuple next() throws PageTupleAccessException {
				// TODO Auto-generated method stub
				if (hasNext()) {
				return getDataTuple(visitedRecordNum);
				} else {
					throw new PageTupleAccessException(visitedRecordNum, "record not available");
				}
			}
			
		};
		
		
	}

	@Override
	public PageTupleIterator getIteratorWithRID()
			throws PageTupleAccessException {
		// TODO Auto-generated method stub
		return null;
	}

	/* read number of records on prage from Header
	 * @see de.tuberlin.dima.minidb.io.tables.TablePage#getNumRecordsOnPage()
	 * @author thuy
	 */
	@Override
	public int getNumRecordsOnPage() {
		return IntField.getIntFromBinary(binaryPage, 8);
	}

	@Override
	public int getPageID() {
		int pageID = 0;
		
		for (int i = 4; i < 8; i++) {
			pageID |= (Integer.reverse(binaryPage[i] & 0xff) >>> (7-i)*8);
		}
		return pageID;
	}

	/*
	 * Vicky
	 */
	/*
	 * return recordWidth with default length of variable columns (BigInt)
	 * (non-Javadoc)
	 * @see de.tuberlin.dima.minidb.io.tables.TablePage#getRecordWidth()
	 */
	@Override
	public int getRecordWidth() {
		int recordWidth=0;
		for (int j = 0; j < 4; j++) {
			recordWidth |= (Integer.reverse(binaryPage[j+12] &0xff ) >>> (3-j)*8);
			}
		return recordWidth;
		
	}

	/*
	 * Vicky
	 */
	@Override
	public boolean hasSpaceAvailable() {
		if (getRecordOffset() < getVariableOffset() - getRecordWidth())
			return true;
		else
			return false;
	}

	/*
	 * (non-Javadoc)
	 * @see de.tuberlin.dima.minidb.io.tables.TablePage#insertDataTuple(de.tuberlin.dima.minidb.core.DataTuple)
	 * @author thuy
	 */
	@Override
	public boolean insertDataTuple(DataTuple tuple) throws PageFormatException {
		
		// current record offset
		int recordOffset = getRecordOffset();
		// metadata
		recordOffset += 4;
		
		if (hasSpaceAvailable()) {
			for (int i = 0; i < tuple.getNumberOfFields(); i++) {
				DataField dataField = tuple.getField(i);
				if (dataField.isFixedLength()) {
					dataField.encodeBinary(binaryPage, recordOffset);
					recordOffset += dataField.getNumberOfBytes();
				} else if (dataField.isArrayType()) {
					
					// calculate BigInt conten
					int fieldLength = dataField.getNumberOfBytes();
					int fieldOffset = getVariableOffset() - fieldLength;
					
					// create and write BigInt
					new IntField(fieldOffset).encodeBinary(binaryPage, recordOffset);
					recordOffset += DataType.INT.getNumberOfBytes();
					new IntField(fieldLength).encodeBinary(binaryPage, recordOffset);
					recordOffset += DataType.INT.getNumberOfBytes();
					
					// write variable chunk
					dataField.encodeBinary(binaryPage, fieldOffset);
					setVariableOffset(fieldOffset);
				} else if (dataField.isNULL()) {
					dataField.encodeBinary(binaryPage, recordOffset);
					recordOffset += dataField.getNumberOfBytes();
				} else {
					throw new PageFormatException("unknown DataFiled Type");
				}
			}
			setNumberOfTuples(getNumRecordsOnPage() + 1);
			return true;
		} else {
			throw new PageFormatException("not enough space");
		}
	}
	
	/*
	 * Vicky
	 */
	private int getRecordOffset() {
		return (int) 32 + getNumRecordsOnPage() * getRecordWidth();
	}

	@Override
	public boolean updateDataTuple(int position, DataTuple newTuple)
			throws PageTupleAccessException, PageFormatException {
		return false;
	}

	@Override
	public boolean hasBeenModified() {
		return false;
	}

}
